package org.learn.dataguru.kafka.demo.class05;

import java.util.Collections;
import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Replica {
	
	private static final int TIMEOUT = 3000;
	
	private static final String prefix = "/kafka";
	
	private static ZooKeeper zkp = null;

	
	public void start() throws Exception{
		ZkUtil util = new ZkUtil();
		zkp = util.getZkp();
		
		//注册
		String currNode = zkp.create(prefix+"/replica", null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		System.out.println("znode:"+currNode);
		
		election(prefix,currNode);
		
		while(true){
			Thread.sleep(1000*5);
		}
	}
	
	/**
	 * 选举
	 * @param currNode
	 * @throws Exception
	 */
	private boolean election(final String prefix,final String currNode) throws Exception{
		String preNode = preNode(prefix,currNode);
		if(preNode==null){
			//没有前节点则当前节点即为leader
			System.out.println("currNode("+currNode+") is leader now");
			System.out.println("执行回调方法");
			return true;
		}else{
			//follower,监听前一个节点
			System.out.println("currNode("+currNode+") is not leader , the preNode is  " + prefix+"/"+preNode);
			zkp.exists(prefix+"/"+preNode, new Watcher(){
				//如果前一个节点删除了，再进行选举
				@Override
				public void process(WatchedEvent event) {
					System.out.println("node:"+prefix+"/"+event.getPath()+",type:"+event.getType());
					if(EventType.NodeDeleted.equals(event.getType())){
						try {
							election(prefix,currNode);
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				}
			});
			return false;
		}
	}
	
	/**
	 * 获取前一个节点
	 * @param prefix
	 * @param node
	 * @return
	 * @throws Exception
	 */
	private  String preNode(String prefix ,String node) throws Exception{
		List<String> nodes = zkp.getChildren(prefix, null);
		Collections.sort(nodes);
		for(int i=0;i<nodes.size();i++){
			System.out.println("node:"+prefix+"/"+nodes.get(i));
			if(i==0 && node.equals(prefix+"/"+nodes.get(i))){
				return null;
			}else if(node.equals(prefix+"/"+nodes.get(i))){
				return nodes.get(i-1);
			}
		}
		throw new Exception("Node does not exist ");
	}

}
